package ar.uba.fi.tonyvaliente.tree;

import java.util.List;
import java.util.logging.Logger;

import ar.uba.fi.tonyvaliente.files.File;
import ar.uba.fi.tonyvaliente.files.FileBlock;
import ar.uba.fi.tonyvaliente.files.FileException;
import ar.uba.fi.tonyvaliente.files.FileRecord;
import ar.uba.fi.tonyvaliente.files.Persistor;
import ar.uba.fi.tonyvaliente.files.PersistorException;
import ar.uba.fi.tonyvaliente.files.RecordPosition;

/**
 * Clase que envuelve un archivo de bloques (File) donde hay un arbol B+
 * y realiza la serializacion y deserializacion de los nodos del mismo.
 * @author Diego Morello
 * @param <K> Clase a la cual deben pertenecer las claves, debe ser comparable
 * @param <V> Clase a la cual deben pertenecer los valores
 */
public class BPlusTreeFileWrapper<K extends Comparable<K>, V> {
	
	private static Logger log = Logger.getLogger(BPlusTreeFileWrapper.class.getName());
	
	private File file;
	private NodePersistor<K,V> nodePersistor;
	
	/**
	 * Crea una nueva instancia de la clase que apunta a un archivo de bloques
	 * donde hay un arbol B+ y sabe como serializar y deserializar las claves y valores del arbol 
	 * @param file Archivo de bloques donde se encuentra el arbol B+
	 * @param keyPersistor Persistor para serializar y deserializar las claves del arbol
	 * @param valuePersistor Persistor para serializar y deserializar los valores del arbol
	 */
	public BPlusTreeFileWrapper(File file, Persistor<K> keyPersistor, Persistor<V> valuePersistor) {
		this.file = file;
		this.nodePersistor = new NodePersistor<K,V>(keyPersistor, valuePersistor, this); 
	}
	
	/**
	 * Cierra el archivo encapsulado
	 * @throws BPlusTreeException si ocurre un error al cerrar el archivo
	 */
	public void close() throws BPlusTreeException {
		try {
			file.close();
		} catch (FileException e) {
			throw new BPlusTreeException("Error al cerrar el BPlusTreeFileWrapper", e);
		}
	}
	
	/**
	 * Obtiene el nodo raiz del arbol. Para ello, busca en el File subyacente las propiedades
	 * rootNodeBlockNumber y rootNodeOffset, que indican el bloque y offset dentro del bloque
	 * donde se encuentra el FileRecord correspondiente al nodo raiz.
	 */
	public RecordPosition getRootNodePosition() throws BPlusTreeException {
		String strRootBlockNumber = file.getProperty("rootNodeBlockNumber");
		String strRootOffset = file.getProperty("rootNodeOffset");
		
		if (strRootBlockNumber != null && strRootOffset != null) {
			try {
				int rootBlockNumber = Integer.parseInt(strRootBlockNumber);
				int rootOffset = Integer.parseInt(strRootOffset);
				return new RecordPosition(rootBlockNumber, rootOffset);
			} catch (NumberFormatException e) {
				throw new BPlusTreeException("Error al obtener posicion del nodo raiz: "
						+ "valor incorrecto", e);
			}
		} else {
			throw new BPlusTreeException("Error en el archivo de arbol: "
					+ "no se encuentra la posicion del nodo raiz");
		}
	}
	
	/**
	 * Asigna al archivo la nueva posicion del nodo raiz
	 * @param rootNode
	 * @throws BPlusTreeException
	 */
	public void setRootNodePosition(RecordPosition pos) throws BPlusTreeException {
		try {
			log.fine("setRootNode IN");
			log.finer("rootNodeBlockNumber: " + pos.getBlockNumber());
			log.finer("rootNodeOffset: " + pos.getOffset());
			
			file.setProperty("rootNodeBlockNumber", Integer.toString(pos.getBlockNumber()));
			file.setProperty("rootNodeOffset", Integer.toString(pos.getOffset()));
		} catch (FileException e) {
			throw new BPlusTreeException("Error al asignar nuevo nodo raiz", e);
		}finally{
			try {
				file.flush();
			} catch (FileException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
	
	/**
	 * Obtiene el nodo cuyo bloque y offset dentro del bloque recibe como parametro.
	 * 
	 * @param blockNumber posicion del bloque dentro del File
	 * @param offset posicion del nodo serializado dentro del bloque
	 * @return nodo recuperado
	 */
	public BPlusTreeNode<K, V> getNode(int blockNumber, int offset) throws BPlusTreeException {
		try {
			log.fine("getNode IN");
			log.finer("BlockNumber: " + blockNumber);
			log.finer("NodeOffset: " + offset);
			
			FileBlock block = file.getBlock(blockNumber);
			FileRecord record = block.getRecord(offset);
			return nodePersistor.deserialize(record.getData());
		} catch (FileException e) {
			throw new BPlusTreeException("Error al acceder al archivo para obtener nodo", e);
		} catch (PersistorException e) {
			throw new BPlusTreeException("Error al deserializar nodo", e);
		}
	}
	
	/**
	 * Agrega un nodo al archivo.
	 * @param node nodo a agregar
	 * @return RecordPosition que contiene el numero de bloque y offset
	 * donde se escribio el nodo
	 */
	public RecordPosition appendNode(BPlusTreeNode<K,V> node) throws BPlusTreeException {
		try {
			
			log.fine("appendNode IN");
			try{
				log.finer("node: " + node);
			}catch(Exception e){
				
			}
			
			FileRecord r = new FileRecord(nodePersistor.serialize(node));
			RecordPosition posicion = file.addRecord(r);
			log.finer("posicion: block: " + posicion.getBlockNumber() + " offset:  " + posicion.getOffset());
			
			return posicion;
		} catch (FileException e) {
			throw new BPlusTreeException("Error al acceder al archivo para agregar nodo", e);
		} catch (PersistorException e) {
			throw new BPlusTreeException("Error al serializar nodo para agregar en archivo", e);
		}
	}
	
	/**
	 * Escribe un nodo en el archivo, en la posicion indicada por parametro
	 * @param node Nodo a actualizar en archivo
	 * @param pos Posicion del archivo donde escribir el nodo
	 * @return TRUE si pudo escribir el nodo, FALSE si el bloque indicado en pos estaba lleno
	 * y no pudo hacerse la actualizacion
	 * @throws BPlusTreeException Si ocurre un error al acceder a archivo o serializar el nodo
	 */
	public boolean updateNode(BPlusTreeNode<K,V> node, RecordPosition pos) throws BPlusTreeException {
		try {
			
			log.fine("updateRootNode IN");
			log.finer("NodeBlockNumber: " + pos.getBlockNumber());
			log.finer("NodeOffset: " + pos.getOffset());
			try{
				log.finer("node: " + node);
			}catch(Exception e){
				
			}
			
			FileBlock block = file.getBlock(pos.getBlockNumber());
			FileRecord r = new FileRecord(nodePersistor.serialize(node));
			boolean pudoActualizar = block.updateRecord(pos.getOffset(), r);
			if (pudoActualizar) {
				file.updateBlock(pos.getBlockNumber(), block);
				return true;
			} else {
				return false;
			}
		} catch (FileException e) {
			throw new BPlusTreeException("Error al acceder a archivo para actualizar nodo", e);
		} catch (PersistorException e) {
			throw new BPlusTreeException("Error al serializar nodo para actualizar en disco", e);
		}
	}
	
	public RecordPosition repositionNode(BPlusTreeNode<K,V> node, RecordPosition originalPos)
			throws BPlusTreeException {
		
		RecordPosition newPos;
		
		int blockNumber = originalPos.getBlockNumber();
		int offset = originalPos.getOffset();
		
		try {
			// Verifico si el nodo a reposicionar es el ultimo del bloque
			FileBlock block = file.getBlock(blockNumber);
			if (offset == block.getRecordCount() - 1) {
				// Borro el nodo del bloque y actualizo el bloque
				block.eraseRecord(offset);
				file.updateBlock(blockNumber, block);
				
				// Agrego el bloque al final del archivo
				newPos = this.appendNode(node);
				
				// Creo una referencia para el nodo reubicado
				Reference<K, V> newRef = node.getTree().createReference();
				DiskReference<K, V> diskNewRef = (DiskReference<K, V>)newRef;
				diskNewRef.setRecordPosition(newPos);
				node.setThisReference(diskNewRef);
				
				// Verifico si es nodo interno u hoja
				if (node.isInner()) {
					// Interno-> actualizo hijos
					InnerBPlusTreeNode<K, V> innerNode;
					innerNode = (InnerBPlusTreeNode<K, V>)node;
					for (Reference<K,V> ref: innerNode.getReferences()) {
						DiskReference<K, V> diskRef = (DiskReference<K, V>)ref;
						BPlusTreeNode<K, V> child = diskRef.getNode();
						child.setParent(node.getThisReference());
						diskRef.setNode(child);
					}
				} else {
					// Hoja -> actualizo nodos siguiente y anterior
					LeafBPlusTreeNode<K, V> leafNode;
					leafNode = (LeafBPlusTreeNode<K, V>)node;
					
					// Actualizo el siguiente
					DiskReference<K, V> next;
					next = (DiskReference<K, V>)leafNode.getNext();
					if (next != null) {
						LeafBPlusTreeNode<K, V> nextNode;
						nextNode = (LeafBPlusTreeNode<K, V>)next.getNode();
						nextNode.setPrevious(node.getThisReference());
						next.setNode(nextNode);
					}
					
					// Actualizo el anterior
					DiskReference<K, V> previous;
					previous = (DiskReference<K, V>)leafNode.getPrevious();
					if (previous != null) {
						LeafBPlusTreeNode<K, V> previousNode;
						previousNode = (LeafBPlusTreeNode<K, V>)previous.getNode();
						previousNode.setNext(node.getThisReference());
						previous.setNode(previousNode);
					}
				}
				
				// Actualizo el nodo padre (si lo tiene)
				DiskReference<K, V> parent;
				parent = (DiskReference<K, V>)node.getParent();
				if (parent != null) {
					InnerBPlusTreeNode<K, V> parentNode;
					parentNode = (InnerBPlusTreeNode<K, V>)parent.getNode();
					for (Reference<K, V> ref: parentNode.getReferences()) {
						DiskReference<K, V> diskRef = (DiskReference<K, V>)ref;
						RecordPosition childPos = diskRef.getRecordPosition();
						if (childPos.equals(originalPos)) {
							diskRef.setRecordPosition(newPos);
							break;
						}
					}
					parent.setNode(parentNode);
				}
				
				// Devuelvo la posicion donde reubique el nodo
			} else {
				// Reubico el ultimo nodo en la posicion originalPos
				// Reubico el nodo en cuestion al final del archivo
				
				FileRecord lastRecord = block.getRecord(block.getRecordCount()-1);
				BPlusTreeNode<K, V> lastNode = nodePersistor.deserialize(lastRecord.getData());
				RecordPosition lastPos = new RecordPosition(blockNumber, block.getRecordCount()-1);
				
				// Si nodoReubicar.padre == ultimoNodo
				DiskReference<K, V> parent = (DiskReference<K, V>)node.getParent();
				if (parent != null && parent.getRecordPosition().equals(lastPos)) {
					// nodoReubicar.padre := originalPos
					parent.setRecordPosition(originalPos);
				} else {
					// si ultimoNodo.padre == nodoReubicar
					parent = (DiskReference<K, V>)lastNode.getParent();
					if (parent != null && parent.getRecordPosition().equals(originalPos)) {
						InnerBPlusTreeNode<K, V> inner = (InnerBPlusTreeNode<K, V>)node;
						List<Reference<K, V>> references = inner.getReferences();
						for (Reference<K, V> ref: references) {
							DiskReference<K, V> diskRef = (DiskReference<K, V>)ref;
							if (diskRef.getRecordPosition().equals(lastPos)) {
								diskRef.setRecordPosition(originalPos);
								break;
							}
						}
					}
				}
				
				// Si nodoReubicar es hoja, veo si lastNode es siguiente o previo
				if (node.isLeaf()) {
					LeafBPlusTreeNode<K, V> leaf = (LeafBPlusTreeNode<K, V>)node;
					DiskReference<K, V> next = (DiskReference<K, V>)leaf.getNext();
					if (next != null && next.getRecordPosition().equals(lastPos)) {
						next.setRecordPosition(originalPos);
					} else {
						DiskReference<K, V> prev = (DiskReference<K, V>)leaf.getPrevious();
						if (prev != null && prev.getRecordPosition().equals(lastPos)) {
							prev.setRecordPosition(originalPos);
						}
					}
				}
				
				file.flush();
				FileRecord newRecord = new FileRecord(nodePersistor.serialize(node));
				FileRecord newLastRecord = new FileRecord(nodePersistor.serialize(lastNode));
				newPos = file.addRecord(newRecord);
				block.eraseRecord(block.getRecordCount()-1);
				block.updateRecord(originalPos.getOffset(), newLastRecord);
				file.updateBlock(originalPos.getBlockNumber(), block);
				
				// Actualizo padre de node
				parent = (DiskReference<K, V>)node.getParent();
				if (parent != null) {
					InnerBPlusTreeNode<K, V> nodeParent;
					nodeParent = (InnerBPlusTreeNode<K, V>)parent.getNode();
					List<Reference<K, V>> parentRefs = nodeParent.getReferences();
					for (Reference<K, V> parentRef: parentRefs) {
						DiskReference<K, V> diskParentRef = (DiskReference<K, V>)parentRef;
						if (diskParentRef.getRecordPosition().equals(originalPos)) {
							diskParentRef.setRecordPosition(newPos);
							break;
						}
					}
					this.updateNode(nodeParent, parent.getRecordPosition());
					DiskReference<K, V> rootRef = (DiskReference<K, V>)node.tree.getRoot();
					if (parent.getRecordPosition().equals(rootRef.getRecordPosition())) {
						node.tree.setRootNode(nodeParent);
					}
				}
				
				
				// Repito para lastNode
				parent = (DiskReference<K, V>)lastNode.getParent();
				if (parent != null) {
					InnerBPlusTreeNode<K, V> nodeParent;
					nodeParent = (InnerBPlusTreeNode<K, V>)parent.getNode();
					List<Reference<K, V>> parentRefs = nodeParent.getReferences();
					for (Reference<K, V> parentRef: parentRefs) {
						DiskReference<K, V> diskParentRef = (DiskReference<K, V>)parentRef;
						if (diskParentRef.getRecordPosition().equals(lastPos)) {
							diskParentRef.setRecordPosition(originalPos);
							break;
						}
					}
					this.updateNode(nodeParent, parent.getRecordPosition());
					DiskReference<K, V> rootRef = (DiskReference<K, V>)node.tree.getRoot();
					if (parent.getRecordPosition().equals(rootRef.getRecordPosition())) {
						node.tree.setRootNode(nodeParent);
					}
				}
				
				// Verifico si node es hoja o interno
				if (node.isInner()) {
					// Actualizo hijos
					InnerBPlusTreeNode<K, V> inner = (InnerBPlusTreeNode<K, V>)node;
					List<Reference<K, V>> references = inner.getReferences();
					for (Reference<K, V> ref: references) {
						BPlusTreeNode<K, V> childNode = ref.getNode();
						DiskReference<K, V> parentRef = (DiskReference<K, V>)childNode.getParent();
						parentRef.setRecordPosition(newPos);
						ref.setNode(childNode);
					}
				} else {
					LeafBPlusTreeNode<K, V> leaf = (LeafBPlusTreeNode<K, V>)node;

					// Actualizo siguiente
					DiskReference<K, V> nextRef = (DiskReference<K, V>)leaf.getNext();
					if (nextRef != null) {
						if (nextRef.getRecordPosition().equals(lastPos)) {
							nextRef.setRecordPosition(originalPos);
						} else {
							LeafBPlusTreeNode<K, V> nextNode;
							nextNode = (LeafBPlusTreeNode<K, V>)nextRef.getNode();
							DiskReference<K, V> nextPrevRef;
							nextPrevRef = (DiskReference<K, V>)nextNode.getPrevious();
							nextPrevRef.setRecordPosition(newPos);
							nextRef.setNode(nextNode);
						}
					}
					
					// Actualizo previo
					DiskReference<K, V> prevRef = (DiskReference<K, V>)leaf.getPrevious();
					if (prevRef != null) {
						if (prevRef.getRecordPosition().equals(lastPos)) {
							prevRef.setRecordPosition(originalPos);
						} else {
							LeafBPlusTreeNode<K, V> prevNode;
							prevNode = (LeafBPlusTreeNode<K, V>)prevRef.getNode();
							DiskReference<K, V> prevNextRef;
							prevNextRef = (DiskReference<K, V>)prevNode.getNext();
							prevNextRef.setRecordPosition(newPos);
							prevRef.setNode(prevNode);
						}
					}
					
				}

				if (lastNode.isInner()) {
					// Actualizo hijos
					InnerBPlusTreeNode<K, V> inner = (InnerBPlusTreeNode<K, V>)lastNode;
					List<Reference<K, V>> references = inner.getReferences();
					for (Reference<K, V> ref: references) {
						BPlusTreeNode<K, V> childNode = ref.getNode();
						DiskReference<K, V> parentRef = (DiskReference<K, V>)childNode.getParent();
						if(parentRef != null){
							// No es la raiz.
							parentRef.setRecordPosition(originalPos);
						}
						ref.setNode(childNode);
					}
				} else {
					LeafBPlusTreeNode<K, V> leaf = (LeafBPlusTreeNode<K, V>)lastNode;

					// Actualizo siguiente
					DiskReference<K, V> nextRef = (DiskReference<K, V>)leaf.getNext();
					if (nextRef != null) {
						if (nextRef.getRecordPosition().equals(originalPos)) {
							nextRef.setRecordPosition(newPos);
						} else {
							LeafBPlusTreeNode<K, V> nextNode;
							nextNode = (LeafBPlusTreeNode<K, V>)nextRef.getNode();
							DiskReference<K, V> nextPrevRef;
							nextPrevRef = (DiskReference<K, V>)nextNode.getPrevious();
							nextPrevRef.setRecordPosition(originalPos);
							nextRef.setNode(nextNode);
						}
					}
					
					// Actualizo previo
					DiskReference<K, V> prevRef = (DiskReference<K, V>)leaf.getPrevious();
					if (prevRef != null) {
						if (prevRef.getRecordPosition().equals(originalPos)) {
							prevRef.setRecordPosition(newPos);
						} else {
							LeafBPlusTreeNode<K, V> prevNode;
							prevNode = (LeafBPlusTreeNode<K, V>)prevRef.getNode();
							DiskReference<K, V> prevNextRef;
							prevNextRef = (DiskReference<K, V>)prevNode.getNext();
							prevNextRef.setRecordPosition(originalPos);
							prevRef.setNode(prevNode);
						}
					}
					this.updateNode(node, newPos);
					this.updateNode(lastNode, originalPos);
				}
			}
			return newPos;
		} catch (FileException e) {
			throw new BPlusTreeException("Error al acceder a archivo para reubicar nodo", e);
		} catch (PersistorException e) {
			throw new BPlusTreeException("Error al serializar nodo para reubicacion", e);
		}
	}
	
	/**
	 * Setea la posición del primer nodo del secuencial set.
	 * 
	 * @param first
	 * @throws BPlusTreeException 
	 */
	public void setFirstNodePosition(RecordPosition first) throws BPlusTreeException{
		try{
			this.file.setProperty("firstNodeBlock", Integer.toString(first.getBlockNumber()));
			this.file.setProperty("firstNodeOffset", Integer.toString(first.getOffset()));
		}catch (FileException e) {
			throw new BPlusTreeException("Error al asignar nuevo first node", e);
		}finally{
			try {
				file.flush();
			} catch (FileException e) {
				e.printStackTrace();
			}
		}
	}
	
	/**
	 * Devuelve la posición del primer nodo del secuencial set.
	 * 
	 * @return
	 * @throws BPlusTreeException
	 */
	public RecordPosition getFirstNodePosition() throws BPlusTreeException {
		String strRootBlockNumber = file.getProperty("firstNodeBlock");
		String strRootOffset = file.getProperty("firstNodeOffset");
		
		if (strRootBlockNumber != null && strRootOffset != null) {
			try {
				int rootBlockNumber = Integer.parseInt(strRootBlockNumber);
				int rootOffset = Integer.parseInt(strRootOffset);
				return new RecordPosition(rootBlockNumber, rootOffset);
			} catch (NumberFormatException e) {
				throw new BPlusTreeException("Error al obtener posicion del first node: "
						+ "valor incorrecto", e);
			}
		} else {
			throw new BPlusTreeException("Error en el archivo de arbol: "
					+ "no se encuentra la posicion del first node");
		}
	}

	public void flush() throws BPlusTreeException {
		try {
			this.file.flush();
		} catch (FileException e) {
			throw new BPlusTreeException("Error al hacer flush", e);
		}
	}
}
